#include "RelOp.h"
#include <string.h>

#if defined __GNUC__ || defined __APPLE__
#include <ext/hash_map>
#else
#include <hash_map>
#endif

#include <map>
#include <vector>

void SelectPipe::Run (Pipe &inPipe, Pipe &outPipe, CNF &selOp, Record &literal)
{
    // should use thread but I think it is not necessary
    ComparisonEngine comp;
    Record temp;
    while(inPipe.Remove(&temp)) {
        if(comp.Compare(&temp, &literal, &selOp)) {
            outPipe.Insert(&temp);
        }
    }
}

void* SelectFileThread (void* arg) {
    SelectFileInfo* mysfi;
    mysfi = (SelectFileInfo*)arg;
    DBFile* inFile = mysfi->inFile;
    Pipe* outPipe = mysfi->outPipe;
    CNF* selOp = mysfi->selOp;
    Record* literal = mysfi->literal;
    int numPages = mysfi->numPages;

    inFile->MoveFirst();

    Record* temp = new Record;
    
    while(inFile->GetNext(*temp, *selOp, *literal) == 1) {
        outPipe->Insert(temp);
    }

    outPipe->ShutDown();
    delete temp;
    delete mysfi;
}

void SelectFile::Run (DBFile &inFile, Pipe &outPipe, CNF &selOp, Record &literal) {
    SelectFileInfo* sfi = new SelectFileInfo();
    sfi->inFile = &inFile;
    sfi->outPipe = &outPipe;
    sfi->selOp = &selOp;
    sfi->literal = &literal;
    sfi->numPages = numPages;

    pthread_create (&thread, NULL, &SelectFileThread, (void*) sfi);
}

void SelectFile::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void SelectFile::Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* ProjectThread(void* arg) {
    ProjectInfo* mypi;
    mypi = (ProjectInfo*) arg;
    Pipe* inPipe = mypi->inPipe;
    Pipe* outPipe = mypi->outPipe;
    int* keepMe = mypi->keepMe;
    int numAttsInput = mypi->numAttsInput;
    int numAttsOutput = mypi->numAttsOutput;
    int numPages = mypi->numPages;

    Record* temp = new Record;
    while(inPipe->Remove(temp)) {
        temp->Project(keepMe, numAttsOutput, numAttsInput);
        outPipe->Insert(temp);
    }

    // need shut down inPipe?
    //inPipe->ShutDown();
    outPipe->ShutDown();

    delete temp;
    delete mypi;
}

void Project :: Run(Pipe &inPipe, Pipe &outPipe, int *keepMe, int numAttsInput, int numAttsOutput) {
    ProjectInfo* pi = new ProjectInfo();
    pi->inPipe = &inPipe;
    pi->outPipe = &outPipe;
    pi->keepMe = keepMe;
    pi->numAttsInput = numAttsInput;
    pi->numAttsOutput = numAttsOutput;
    pi->numPages = numPages;

    pthread_create(&thread, NULL, &ProjectThread, (void*) pi);
}

void Project::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void Project::Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* JoinThread(void* arg) {
    JoinInfo* myji;
    myji = (JoinInfo*) arg;
    Pipe* inPipeL = myji->inPipeL;
    Pipe* inPipeR = myji->inPipeR;
    Pipe* outPipe = myji->outPipe;
    CNF* selOp = myji->selOp;
    Record* literal = myji->literal;        // what is this used for?
    int numPages = myji->numPages;

    OrderMaker* leftOrder = new OrderMaker;
    OrderMaker* rightOrder = new OrderMaker;

    int correctOrder = selOp->GetSortOrders(*leftOrder, *rightOrder);

    if(correctOrder == 1) {
        
        int pipesz = 100;
        Pipe* outPipeL = new Pipe(pipesz);
        Pipe* outPipeR = new Pipe(pipesz);

        Record* left_temp = new Record;
        Record* right_temp = new Record;

        BigQ bigQL (*inPipeL, *outPipeL, *leftOrder, numPages);
        BigQ bigQR (*inPipeR, *outPipeR, *rightOrder, numPages);

        ComparisonEngine comp;
        bool isEmpty = false;

        outPipeL->Remove(left_temp);
        outPipeR->Remove(right_temp);

        int numAttsLeft = ((int*)left_temp->bits)[1]/sizeof(int)-1;
        int numAttsRight = ((int*)right_temp->bits)[1]/sizeof(int)-1;
        int numAttsToKeep = numAttsLeft + numAttsRight;
        int* attsToKeep = new int[numAttsToKeep];
        int startOfRight = numAttsLeft;

        for(int i = 0; i < numAttsLeft; i++) {
            attsToKeep[i] = i;
        }
        for(int i = 0; i < numAttsRight; i++) {
            attsToKeep[i + numAttsLeft] = i;
        }

        Record* tempRec = new Record;
        vector<Record*> leftMerge;
        vector<Record*> rightMerge;
        while(!isEmpty) {
            if(comp.Compare(left_temp, leftOrder, right_temp, rightOrder) < 0) {
                outPipeL->Remove(left_temp) == 1 ? isEmpty = false : isEmpty = true;
            }
            else if(comp.Compare(left_temp, leftOrder, right_temp, rightOrder) == 0) {
                Record* leftRec = new Record;
                Record* rightRec = new Record;
                leftRec->Consume(left_temp);
                rightRec->Consume(right_temp);
                leftMerge.push_back(leftRec);
                rightMerge.push_back(rightRec);

                while(true) {
                    if(outPipeL->Remove(left_temp) == 1) {
                        if(comp.Compare(left_temp, leftMerge.back(), leftOrder) == 0) {
                            leftRec = new Record;
                            leftRec->Consume(left_temp);
                            leftMerge.push_back(left_temp);
                        }
                        else {
                            break;
                        }
                    }
                    else {
                        isEmpty = true;
                        break;
                    }
                }
                while(true) {
                    if(outPipeR->Remove(right_temp) == 1) {
                         if(comp.Compare(right_temp, rightMerge.back(), rightOrder) == 0) {
                            rightRec = new Record;
                            rightRec->Consume(right_temp);
                            rightMerge.push_back(rightRec);
                        }
                        else {
                            break;
                        }
                    }
                    else {
                        isEmpty = true;
                        break;
                    }
                }

                for(int i = 0; i < leftMerge.size(); i++) {
                    for(int j = 0; j < rightMerge.size(); j++) {
                        tempRec->MergeRecords(leftMerge.at(i), rightMerge.at(j), numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, startOfRight);
                        outPipe->Insert(tempRec);
                    }
                }

                for(int i = 0; i < leftMerge.size(); i++) {
                    delete leftMerge.at(i);
                }
                leftMerge.clear();
                for(int i = 0; i < rightMerge.size(); i++) {
                    delete rightMerge.at(i);
                }
                rightMerge.clear();
            }
            else {
                outPipeR->Remove(right_temp) == 1 ? isEmpty = false : isEmpty = true;
            }
        }

        delete left_temp;
        delete right_temp;
        delete [] attsToKeep;
        delete tempRec;
        outPipeL->ShutDown();
        outPipeR->ShutDown();
        delete outPipeL;
        delete outPipeR;
    }
    else {
        // default mode: block-nested loop join
       
        Record* left_temp = new Record;
        Record* right_temp = new Record;

        Record* left_rec = new Record;
        Record* right_rec = new Record;

        vector<Record*> left_block;
        vector<Record*> right_block;

        ComparisonEngine comp;

        inPipeL->Remove(left_temp);
        left_rec->Consume(left_temp);
        left_block.push_back(left_rec);
        inPipeR->Remove(right_temp);
        right_rec->Consume(right_temp);
        right_block.push_back(right_rec);

        int numAttsLeft = ((int*)left_rec->bits)[1]/sizeof(int)-1;
        int numAttsRight = ((int*)right_rec->bits)[1]/sizeof(int)-1;
        int numAttsToKeep = numAttsLeft + numAttsRight;
        int* attsToKeep = new int[numAttsToKeep];
        int startOfRight = numAttsLeft;

        for(int i = 0; i < numAttsLeft; i++) {
            attsToKeep[i] = i;
        }
        for(int i = 0; i < numAttsRight; i++) {
            attsToKeep[i + numAttsLeft] = i;
        }

        while(inPipeL->Remove(left_temp)) {
            left_rec = new Record;
            left_rec->Consume(left_temp);
            left_block.push_back(left_rec);
        }
        while(inPipeR->Remove(right_temp)) {
            right_rec = new Record;
            right_rec->Consume(right_temp);
            right_block.push_back(right_rec);
        }

        for(int i = 0; i < left_block.size(); i++) {
            for(int j = 0; j < right_block.size(); j++) {
                if(comp.Compare(left_block.at(i), leftOrder, right_block.at(j), rightOrder) == 0) {
                    Record* tempRec = new Record;
                    tempRec->MergeRecords(left_block.at(i), right_block.at(j), numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, startOfRight);
                    outPipe->Insert(tempRec);
                    delete tempRec;
                }
            }
        }

        for(int i = 0; i < left_block.size(); i++) {
            delete left_block.at(i);
        }
        for(int i = 0; i < right_block.size(); i++) {
            delete right_block.at(i);
        }
        delete left_temp;
        delete right_temp;

    }

    // need to shut down inPipes?
    outPipe->ShutDown();
    delete leftOrder;
    delete rightOrder;
    delete myji;
}

void Join :: Run(Pipe &inPipeL, Pipe &inPipeR, Pipe &outPipe, CNF &selOp, Record &literal) {
    JoinInfo* ji = new JoinInfo();
    ji->inPipeL = &inPipeL;
    ji->inPipeR = &inPipeR;
    ji->outPipe = &outPipe;
    ji->selOp = &selOp;
    ji->literal = &literal;
    ji->numPages = numPages;

    pthread_create(&thread, NULL, &JoinThread, (void*) ji);
}

void Join::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void Join::Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* DuplicateRemovalThread(void* arg) {
    DupRemovalInfo* mydri;
    mydri = (DupRemovalInfo*) arg;
    Pipe* inPipe = mydri->inPipe;
    Pipe* outPipe = mydri->outPipe;
    Schema* myScheam = mydri->mySchema;
    int numPages = mydri->numPages;

    OrderMaker* myOrderMaker = new OrderMaker(myScheam);

    typedef map<Record*, int, HashEqStr> my_hash;
    my_hash recs((HashEqStr(*myOrderMaker)));

    Record* temp = new Record;
    typedef pair <Record*, int> recPair;
    while(inPipe->Remove(temp)) {
        if(recs.find(temp) == recs.end()) {
            Record* r = new Record;
            // expensive copy
            // Insert would zero out the record
            r->Copy(temp);
            recs.insert(recPair (r, 1));
            outPipe->Insert(temp);
        }
    }

    // need to shut down inPipe?
    //inPipe->ShutDown();
    outPipe->ShutDown();
    for(my_hash::iterator it = recs.begin(); it != recs.end(); ++it) {
        delete it->first;
    }
    delete myOrderMaker;
    delete temp;
    delete mydri;
}

void DuplicateRemoval :: Run(Pipe &inPipe, Pipe &outPipe, Schema &mySchema) {
    DupRemovalInfo* dri = new DupRemovalInfo();
    dri->inPipe = &inPipe;
    dri->outPipe = &outPipe;
    dri->mySchema = &mySchema;
    dri->numPages = numPages;

    pthread_create(&thread, NULL, &DuplicateRemovalThread, (void*) dri);
}

void DuplicateRemoval::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void DuplicateRemoval::Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* SumThread(void* arg) {
    SumInfo* mysi;
    mysi = (SumInfo*) arg;
    Pipe* inPipe = mysi->inPipe;
    Pipe* outPipe = mysi->outPipe;
    Function computeMe = mysi->computeMe;
    int numPages= mysi->numPages;

    Record* temp = new Record;
    Type t;
    int intResult = 0;
    int intTemp = 0;
    double doubleResult =0;
    double doubleTemp = 0;
    while(inPipe->Remove(temp)) {
        t = computeMe.Apply(*temp, intTemp, doubleTemp);
        if( t == Int) intResult += intTemp;
        else doubleResult += doubleTemp;
    }

    temp->CreateRecord(t, intResult, doubleResult);
    outPipe->Insert(temp);
    outPipe->ShutDown();
    delete temp;
    delete mysi;
}

void Sum :: Run(Pipe &inPipe, Pipe &outPipe, Function &computeMe) {
    SumInfo* si = new SumInfo();
    si->inPipe = &inPipe;
    si->outPipe = &outPipe;
    si->computeMe = computeMe;
    si->numPages = numPages;

    pthread_create(&thread, NULL, &SumThread, (void*) si);
}

void Sum :: WaitUntilDone () {
    pthread_join (thread, NULL);
}

void Sum :: Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* GroupByThread(void* arg) {
    GroupByInfo* mygbi;
    mygbi = (GroupByInfo*) arg;
    Pipe* inPipe = mygbi->inPipe;
    Pipe* outPipe = mygbi->outPipe;
    OrderMaker* groupAtts = mygbi->groupAtts;
    Function computeMe = mygbi->computeMe;
    int numPages = mygbi->numPages;

    int intResult = 0;
    int intTemp = 0;
    double doubleResult =0;
    double doubleTemp = 0;

    int pipesz = 100;
    Pipe* bigqOutPipe = new Pipe(pipesz);

    BigQ bigq (*inPipe, *bigqOutPipe, *groupAtts, numPages);

    vector<Record*> group;

    Record* temp = new Record;
    Record* tempRec = new Record;
    Record* rec = new Record;
    ComparisonEngine comp;
    bigqOutPipe->Remove(temp);
    rec->Consume(temp);
    group.push_back(rec);

    int numAttsLeft = 1;
    int numAttsRight = groupAtts->numAtts;
    int* attsToKeep = new int[numAttsLeft + numAttsRight];
    attsToKeep[0] = 0;
    int numAttsToKeep = numAttsLeft + numAttsRight;
    int startOfRight = 1;
    for(int i = 0; i < numAttsRight; i++) {
        attsToKeep[i + 1] = groupAtts->whichAtts[i];
    }

    while(bigqOutPipe->Remove(temp)) { //
        if(comp.Compare(temp, group.back(), groupAtts) == 0) { //
            rec = new Record;
            rec->Consume(temp); //
            group.push_back(rec);
        }
        else {
            Type t;
            intResult = 0;
            doubleResult = 0;
            for(int i = 0; i < group.size(); i++) {
                t = computeMe.Apply(*group.at(i), intTemp, doubleTemp);
                if( t == Int) intResult += intTemp;
                else doubleResult += doubleTemp;
                //outPipe->Insert(group.at(i));
            }
            
            Record* sum = new Record;
            sum->CreateRecord(t, intResult, doubleResult);

            tempRec->MergeRecords(sum, group.at(0), numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, startOfRight);
            outPipe->Insert(tempRec);

            rec = new Record;
            rec->Consume(temp); //

            //
            for(int i = 0; i < group.size(); i++) {
                delete group.at(i);
            }
            group.clear();
            group.push_back(rec);
            delete sum;
        }
    }
    // for the last group
    Type t;
    intResult = 0;
    doubleResult = 0;
    for(int i = 0; i < group.size(); i++) {
        t = computeMe.Apply(*group.at(i), intTemp, doubleTemp);
        if( t == Int) intResult += intTemp;
        else doubleResult += doubleTemp;
        //outPipe->Insert(group.at(i));
    }

    Record* sum = new Record;
    sum->CreateRecord(t, intResult, doubleResult);

    tempRec->MergeRecords(sum, group.at(0), numAttsLeft, numAttsRight, attsToKeep, numAttsToKeep, startOfRight);
    outPipe->Insert(tempRec);

    for(int i = 0; i < group.size(); i++) {
        delete group.at(i);
    }
    group.clear();
    delete sum;

    // need to shut down inPipe?
    outPipe->ShutDown();
    delete bigqOutPipe;
    delete [] attsToKeep;
    delete temp;
    delete tempRec;
}

void GroupBy :: Run(Pipe &inPipe, Pipe &outPipe, OrderMaker &groupAtts, Function &computeMe) {
    GroupByInfo* gbi = new GroupByInfo();
    gbi->inPipe = &inPipe;
    gbi->outPipe = &outPipe;
    gbi->groupAtts = &groupAtts;
    gbi->computeMe = computeMe;
    gbi->numPages = numPages;

    pthread_create(&thread, NULL, &GroupByThread, (void*) gbi);
}

void GroupBy::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void GroupBy::Use_n_Pages (int runlen) {
    numPages = runlen;
}

void* WriteOutThread(void* arg) {
    WriteOutInfo* mywoi;
    mywoi = (WriteOutInfo*) arg;
    Pipe* inPipe = mywoi->inPipe;
    FILE* outFile = mywoi->outFile;
    Schema* mySchema = mywoi->mySchema;
    int numPages = mywoi->numPages;

    Record* temp = new Record;

    while(inPipe->Remove(temp)) {
        temp->Write(outFile, mySchema);
    }

    // need to shut down inPipe?
    //inPipe->ShutDown();
    delete temp;
    delete mywoi;
}

void WriteOut :: Run(Pipe &inPipe, FILE *outFile, Schema &mySchema) {
    WriteOutInfo* woi = new WriteOutInfo();
    woi->inPipe = &inPipe;
    woi->outFile = outFile;
    woi->mySchema = &mySchema;
    woi->numPages = numPages;

    pthread_create(&thread, NULL, &WriteOutThread, (void*) woi);
}

void WriteOut::WaitUntilDone () {
    pthread_join (thread, NULL);
}

void WriteOut::Use_n_Pages (int runlen) {
    numPages = runlen;
}
